package delivery

import (
	"github.com/jmoiron/sqlx"
	"learning/gooop/saga/mqs/database"
	"learning/gooop/saga/mqs/models"
	"time"
)

type tDBMsgSource struct {
	clientID string
	expireFunc tSourceExpireFunc
	msgChan chan *models.QueuedMsg
}

func newDBMsgSource(clientID string, expireFunc tSourceExpireFunc) iMsgSource {
	it := new(tDBMsgSource)
	it.init(clientID, expireFunc)
	return it
}

func (me *tDBMsgSource) init(clientID string, expireFunc tSourceExpireFunc) {
	me.clientID = clientID
	me.expireFunc = expireFunc
	me.msgChan = make(chan *models.QueuedMsg, 1)
	go me.beginPollDB()
}

func (me *tDBMsgSource) MsgChan() <- chan *models.QueuedMsg {
	return me.msgChan
}

func (me *tDBMsgSource) beginPollDB() {
	interval := time.Duration(1) * time.Second
	for !me.expireFunc() {
		if len(me.msgChan) <= 0 {
			ok, msg := me.poll()
			if ok {
				me.msgChan <- msg
				continue
			}
		}

		// poll failed, or chan full
		time.Sleep(interval)
	}

	close(me.msgChan)
}

func (me *tDBMsgSource) poll() (bool, *models.QueuedMsg) {
	msg := &models.QueuedMsg{}
	e := database.DB(func(db *sqlx.DB) error {
		rows, err := db.Queryx(
			"select * from delivery_queue where ClientID=? order by CreateTime asc limit 1",
			me.clientID,
		)
		if err != nil {
			return err
		}

		defer rows.Close()
		if rows.Next() {
			err = rows.StructScan(msg)
			if err != nil {
				return err
			}
			return nil

		} else {
			return gEmptyRowsErr
		}
	})

	if e != nil {
		return false, nil
	} else {
		return true, msg
	}
}